package com.dx.architectureddd.common.dp.pipeline;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.util.CollectionUtils;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;

/**
 * 管道执行器
 *
 * @author xin.deng
 * @date 2023/09/24
 */
@Slf4j
public class PipelineFactory<E extends PipelineContext> implements InitializingBean, ApplicationContextAware {

    private ApplicationContext applicationContext;
    private final Map<Class<E>, List<ContextHandler<E>>> pipelineRouteMap = new HashMap<>();
    /**
     * 管道线程池
     */
    private ThreadPoolExecutor pipelineThreadPool;
    private final Class<E> e;
    private final List<Class<ContextHandler<E>>> classList;

    public PipelineFactory(Class<E> e, List<Class<ContextHandler<E>>> classList) {
        this.e = e;
        this.classList = classList;
    }

    /**
     * 同步处理输入的上下文数据
     * 如果处理时上下文数据流通到最后一个处理器且最后一个处理器返回 true，则返回 true，否则返回 false
     *
     * @param context 输入的上下文数据
     * @return 处理过程中管道是否畅通，畅通返回 true，不畅通返回 false
     */
    public boolean acceptSync(E context) {
        Objects.requireNonNull(context, "上下文数据不能为 null");
        // 拿到数据类型
        Class<? extends PipelineContext> dataType = context.getClass();
        // 获取数据处理管道
        List<? extends ContextHandler<E>> pipeline = pipelineRouteMap.get(dataType);

        if (CollectionUtils.isEmpty(pipeline)) {
            log.error("{} 的管道为空", dataType.getSimpleName());
            return false;
        }

        // 管道是否畅通
        boolean lastSuccess = true;

        for (ContextHandler<E> handler : pipeline) {
            try {
                // 当前处理器处理数据，并返回是否继续向下处理
                lastSuccess = handler.handle(context);
            } catch (Throwable ex) {
                log.error("[{}] 处理异常，handler={}", context.getName(), handler.getClass().getSimpleName(), ex);
                throw ex;
            }

            // 不再向下处理
            if (!lastSuccess) {
                break;
            }
        }

        return lastSuccess;
    }

    /**
     * 异步处理输入的上下文数据
     *
     * @param context  上下文数据
     * @param callback 处理完成的回调
     */
    public void acceptAsync(E context, BiConsumer<PipelineContext, Boolean> callback) {
        pipelineThreadPool.execute(() -> {
            boolean success = acceptSync(context);

            if (callback != null) {
                callback.accept(context, success);
            }
        });
    }

    @Override
    public void afterPropertiesSet() {
        this.pipelineThreadPool = applicationContext.getBean(ThreadPoolExecutor.class);
        this.pipelineRouteMap.put(e, this.toPipeline(classList));
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    /**
     * 根据给定的管道中 ContextHandler 的类型的列表，构建管道
     * todo jdk需要到1.8.0_261
     */
    private List<ContextHandler<E>> toPipeline(List<Class<ContextHandler<E>>> classList) {
        return classList.stream()
                .map(applicationContext::getBean)
                .collect(Collectors.toList());
    }
}
